package com.v5ba.server.service.impl;

import com.v5ba.server.common.Constant;
import com.v5ba.server.service.IndexService;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;

import java.io.UnsupportedEncodingException;
import java.util.Date;

/**
 * @author yiren
 * @date 2018/6/17
 */
@Service
public class IndexServiceImpl implements IndexService {


    @Autowired
    @Qualifier(value = "myDefaultMQProducer")
    private DefaultMQProducer defaultMQProducer;

    @Autowired
    @Qualifier("myTransactionMQProducer")
    private TransactionMQProducer transactionMQProducer;



    @Override
    public void exec() throws UnsupportedEncodingException {
        byte[] strByte = ("普通生产者方法演示"+new Date().getTime()).getBytes(RemotingHelper.DEFAULT_CHARSET);
        Message message = new Message(Constant.topic,"tagsA-test", strByte);
        try {
            defaultMQProducer.send(message,new SendCallback(){
                @Override
                public void onSuccess(SendResult sendResult) {
                    System.out.println("exec 处理成功");
                }

                @Override
                public void onException(Throwable e) {
                    System.out.println("exec 处理失败");
                }
            });

        } catch (MQClientException e) {
            e.printStackTrace();
        } catch (RemotingException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     *
     * @throws UnsupportedEncodingException
     */
    public void exec2() throws UnsupportedEncodingException {
        byte[] strByte = ("事物生产者方法演示"+new Date().getTime()).getBytes(RemotingHelper.DEFAULT_CHARSET);

        try {
            transactionMQProducer.setTransactionListener(new TransactionListener(){
                @Override
                public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                    System.out.println("TransactionListener.executeLocalTransaction arg="+arg);

                    if ("addSysLog".equals(arg)){
                        return LocalTransactionState.COMMIT_MESSAGE;
                    }else if("addUser".equals(arg)){
                        return LocalTransactionState.COMMIT_MESSAGE;
                    }

                    return LocalTransactionState.UNKNOW;
                }

                @Override
                public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                    System.out.println("测试一下：TransactionListener.checkLocalTransaction");
                    return LocalTransactionState.COMMIT_MESSAGE;
                }
            });
            Message addSysLogMsg = new Message(Constant.topicOrder,"tagsA-log-tran", strByte);
            TransactionSendResult sendResult = transactionMQProducer.sendMessageInTransaction(addSysLogMsg,"addSysLog");
            System.out.println("发送第一条MQ："+sendResult);


            Message addUserMsg = new Message(Constant.topicOrder,"tagsA-user-tran", strByte);
            TransactionSendResult sendResult2 = transactionMQProducer.sendMessageInTransaction(addUserMsg,"addUser");
            System.out.println("发送第二条MQ："+sendResult2);
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }
}
